package com.wngbms.async;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.EventListener;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;

@Slf4j
public class CompletableFutureDemo {

    public static class M {

        private double price;
        private String name;

        public M(double price, String name) {
            this.price = price;
            this.name = name;
        }

        public double getPrice() {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return price;
        }

        public String getName() {
            return name;
        }

        @Override
        public String toString() {
            return "M{" +
                    "price=" + price +
                    ", name='" + name + '\'' +
                    '}';
        }
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
//        m1();
//        m2();
//        m3();
//        m4();
//        m5();
//        m6();
//        m7();
//        m8();
//        m9();


    }

    private static void m9() {
        CompletableFuture<Integer> f1 = CompletableFuture.supplyAsync(() -> {
            log.info("do run 1");
            return 1;
        });

        CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync(() -> {
            log.info("do run 2");
            return 2;
        });

        CompletableFuture<Integer> r = f1.thenCombine(f2, (x, y) -> {
            log.info("combine");
            return x + y;
        });

        log.info("{}", r.join());
    }

    private static void m8() {
        CompletableFuture<String> p1 = CompletableFuture.supplyAsync(() -> {
            sleep();
            sleep();
            log.info("do run 1");
            return "01";
        });

        CompletableFuture<String> p2 = CompletableFuture.supplyAsync(() -> {
            sleep();
            log.info("do run 2");
            return "02";
        });

        CompletableFuture<String> r = p1.applyToEither(p2, f -> {
            log.info("do run r");
            return f + " win";
        });

        log.info("{}", r.join());
    }

    private static void m7() {
        CompletableFuture
                .runAsync(() -> {
                    log.info("do run"); //  [ForkJoinPool.commonPool-worker-25]
                })
                .thenRun(() -> {
                    log.info("do run 1"); //  [ForkJoinPool.commonPool-worker-25
                })
                .thenRun(() -> {
                    log.info("do run 2"); //  [ForkJoinPool.commonPool-worker-25
                })
                .thenRun(() -> {
                    log.info("do run 3"); //  [ForkJoinPool.commonPool-worker-25
                })
                .join();

        log.info("-----------------------------------------------1");

        CompletableFuture
                .runAsync(() -> {
                    log.info("do run"); //  [ForkJoinPool.commonPool-worker-25]
                })
                .thenRunAsync(() -> {
                    log.info("do run 1"); //  [ForkJoinPool.commonPool-worker-25
                })
                .thenRunAsync(() -> {
                    log.info("do run 2"); //  [ForkJoinPool.commonPool-worker-25
                })
                .thenRunAsync(() -> {
                    log.info("do run 3"); //  [ForkJoinPool.commonPool-worker-25
                })
                .join();

        log.info("-----------------------------------------------2");

        ExecutorService service = Executors.newFixedThreadPool(3);
        CompletableFuture
                .runAsync(() -> {
                    log.info("do run"); // [pool-1-thread-1]
                }, service)
                .thenRun(() -> {
                    log.info("do run 1"); // [pool-1-thread-1]
                })
                .thenRun(() -> {
                    log.info("do run 2"); // [main]
                })
                .thenRun(() -> {
                    log.info("do run 3"); // [main]
                })
                .join();

        log.info("-----------------------------------------------3");

        CompletableFuture
                .runAsync(() -> {
                    sleep();
                    log.info("do run"); // [pool-1-thread-2]
                }, service)
                .thenRun(() -> {
                    sleep();
                    log.info("do run 1"); // [pool-1-thread-2]
                })
                .thenRun(() -> {
                    sleep();
                    log.info("do run 2"); // [pool-1-thread-2]
                })
                .thenRun(() -> {
                    sleep();
                    log.info("do run 3"); // [pool-1-thread-2]
                })
                .join();

        log.info("-----------------------------------------------4");

        CompletableFuture
                .supplyAsync(() -> {
                    log.info("do run"); // [pool-1-thread-3]
                    return 1;
                }, service)
                .thenRunAsync(() -> {
                    sleep();
                    log.info("do run 1"); // [ForkJoinPool.commonPool-worker-25]
                })
                .thenRunAsync(() -> {
                    sleep();
                    log.info("do run 2"); // [pool-1-thread-1]
                }, service)
                .thenRunAsync(() -> {
                    sleep();
                    log.info("do run 3"); // [ForkJoinPool.commonPool-worker-25]
                })
                .join();

        log.info("-----------------------------------------------5");

        CompletableFuture
                .supplyAsync(() -> {
                    log.info("do run"); // [pool-1-thread-2]
                    return 1;
                }, service)
                .thenRunAsync(() -> {
                    log.info("do run 1"); // [ForkJoinPool.commonPool-worker-25]
                })
                .thenRunAsync(() -> {
                    log.info("do run 2"); // [pool-1-thread-3]
                }, service)
                .thenRunAsync(() -> {
                    log.info("do run 3"); // [ForkJoinPool.commonPool-worker-25]
                })
                .join();

        service.shutdown();
    }

    private static void sleep() {
        try {
            TimeUnit.MICROSECONDS.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static void m6() {
        log.info("{}", CompletableFuture
                .supplyAsync(() -> {
                    log.info("supply");
                    return "windows";
                })
                .thenRun(() -> log.info("do thenRun"))
                .join()); // main

        log.info("{}", CompletableFuture
                .supplyAsync(() -> {
                    log.info("supply");
                    return "mac os";
                })
                .thenAccept(r -> log.info("do thenAccept {}", r)) // main
                .join());  // main

        log.info("{}", CompletableFuture
                .supplyAsync(() -> {
                    log.info("supply");
                    return "ipad os";
                })
                .thenApply(r -> {
                    log.info("do thenApply {}", r); // main
                    return "ipad pro";
                })
                .join()); // main

    }

    private static void m5() {
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        CompletableFuture
                .supplyAsync(() -> {
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return 1;
                }, executorService)
                .handle((f, e) -> {
                    log.info("stage 1 {} {}", f, e != null ? e.getMessage() : null);
                    // 有异常 thenApply会中断
                    int i = 10 / 0;
                    return f + 2;
                })
                .handle((f, e) -> {
                    log.info("stage 2 {} {}", f, e != null ? e.getMessage() : null);
                    return f + 3;
                })
                .whenComplete((v, e) -> {
                    log.info("res: {} {}", v, e != null ? e.getMessage() : null);
                })
                .exceptionally(e -> {
                    log.error("exp", e != null ? e.getMessage() : null);
                    return null;
                });
        log.info("main end");
        executorService.shutdown();
    }

    private static void m4() {
        ExecutorService executorService = Executors.newFixedThreadPool(3);

        CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 1;
        }, executorService).thenApply(f -> {
            log.info("stage 1");
            // 有异常 thenApply会中断
            int i = 10 / 0;
            return f + 2;
        }).thenApply(f -> {
            log.info("stage 2");
            return f + 3;
        }).whenComplete((v, e) -> {
            log.info("res: {} {}", v, e);
        }).exceptionally(e -> {
            log.error("", e);
            return null;
        });

        log.info("main end");
    }

    private static void m3() {
        List<M> l = new ArrayList<>();
        l.add(new M(234.2, "miui"));
        l.add(new M(211.2, "sony"));
        l.add(new M(255.2, "apple"));

        List<Double> l2 = l.stream()
                .map(x -> CompletableFuture.supplyAsync(() -> {
                    log.info("x:{}", x);
                    return x.getPrice();
                }))
                .collect(Collectors.toList())
                .stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList());
        log.info("l2:{}", l2);
    }

    private static void m2() {
        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello";
        });

        // join 不会抛出异常
//        System.out.println(completableFuture.join());
        try {
            TimeUnit.SECONDS.sleep(4);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(completableFuture.complete("cv") + "\t" + completableFuture.join());

    }

    private static void m1() {
        ExecutorService executor = Executors.newFixedThreadPool(3);

        try {
            CompletableFuture.supplyAsync(() -> {
                log.info("exec");
                try {
                    TimeUnit.SECONDS.sleep(3);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return 123;
            }, executor).whenComplete((v, e) -> {
                log.info("do handle", v);
            }).exceptionally(e -> {
                log.error("exp", e);
                return null;
            });

            log.info("main exec");
        } finally {
            executor.shutdown();
        }
    }

}
